#!/bin/bash

# 测试smart_tour数据库Maxwell监控脚本
# 用于验证Maxwell是否能正确捕捉smart_tour数据库的变化

echo "=== 测试smart_tour数据库Maxwell监控 ==="
echo "时间: $(date)"
echo ""

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

# 配置变量
MYSQL_HOST="192.168.10.102"
MYSQL_PORT="3306"
MYSQL_USER="root"
MYSQL_PASSWORD="000000"
MAXWELL_USER="maxwell"
MAXWELL_PASSWORD="maxwell"
DATABASE="smart_tour"
KAFKA_TOPIC="maxwell_smart_tour"

# 检查smart_tour数据库
check_smart_tour_database() {
    echo -e "${BLUE}1. 检查smart_tour数据库${NC}"
    echo "----------------------------------------"
    
    # 检查数据库是否存在
    if mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD -e "USE $DATABASE;" 2>/dev/null; then
        echo -e "${GREEN}✓ smart_tour数据库存在${NC}"
        
        # 显示数据库中的表
        echo "数据库中的表:"
        mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD -e "USE $DATABASE; SHOW TABLES;" 2>/dev/null
        
        return 0
    else
        echo -e "${RED}✗ smart_tour数据库不存在${NC}"
        echo "请先创建smart_tour数据库"
        return 1
    fi
}

# 检查maxwell用户权限
check_maxwell_permissions() {
    echo -e "${BLUE}2. 检查maxwell用户权限${NC}"
    echo "----------------------------------------"
    
    if mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MAXWELL_USER -p$MAXWELL_PASSWORD -e "SELECT 1;" 2>/dev/null; then
        echo -e "${GREEN}✓ Maxwell用户连接成功${NC}"
        
        # 检查对smart_tour数据库的权限
        echo "检查对smart_tour数据库的权限..."
        if mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MAXWELL_USER -p$MAXWELL_PASSWORD -e "USE $DATABASE; SELECT 1;" 2>/dev/null; then
            echo -e "${GREEN}✓ Maxwell用户对smart_tour数据库有访问权限${NC}"
        else
            echo -e "${RED}✗ Maxwell用户对smart_tour数据库无访问权限${NC}"
            return 1
        fi
        
        return 0
    else
        echo -e "${RED}✗ Maxwell用户连接失败${NC}"
        return 1
    fi
}

# 创建测试表（如果不存在）
create_test_table() {
    echo -e "${BLUE}3. 创建测试表${NC}"
    echo "----------------------------------------"
    
    echo "创建测试表..."
    mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;

-- 创建测试表（如果不存在）
CREATE TABLE IF NOT EXISTS maxwell_test_table (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100),
    description TEXT,
    status ENUM('active', 'inactive') DEFAULT 'active',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 插入一些初始数据
INSERT IGNORE INTO maxwell_test_table (name, description) VALUES 
('测试项目1', '这是第一个测试项目'),
('测试项目2', '这是第二个测试项目'),
('测试项目3', '这是第三个测试项目');

-- 显示表结构
DESCRIBE maxwell_test_table;

-- 显示数据
SELECT * FROM maxwell_test_table;
EOF
    
    if [ $? -eq 0 ]; then
        echo -e "${GREEN}✓ 测试表创建成功${NC}"
        return 0
    else
        echo -e "${RED}✗ 测试表创建失败${NC}"
        return 1
    fi
}

# 检查Kafka主题
check_kafka_topic() {
    echo -e "${BLUE}4. 检查Kafka主题${NC}"
    echo "----------------------------------------"
    
    if command -v kafka-topics.sh >/dev/null 2>&1; then
        # 检查主题是否存在
        if kafka-topics.sh --bootstrap-server hadoop102:9092 --list 2>/dev/null | grep -q "$KAFKA_TOPIC"; then
            echo -e "${GREEN}✓ Kafka主题 '$KAFKA_TOPIC' 存在${NC}"
            
            # 显示主题详情
            echo "主题详情:"
            kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic $KAFKA_TOPIC 2>/dev/null
        else
            echo -e "${YELLOW}⚠ Kafka主题 '$KAFKA_TOPIC' 不存在，正在创建...${NC}"
            kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic $KAFKA_TOPIC --partitions 3 --replication-factor 1 2>/dev/null
            if [ $? -eq 0 ]; then
                echo -e "${GREEN}✓ Kafka主题创建成功${NC}"
            else
                echo -e "${RED}✗ Kafka主题创建失败${NC}"
                return 1
            fi
        fi
    else
        echo -e "${YELLOW}⚠ 未找到kafka-topics.sh，跳过Kafka检查${NC}"
    fi
    
    return 0
}

# 启动Maxwell监控
start_maxwell_monitoring() {
    echo -e "${BLUE}5. 启动Maxwell监控${NC}"
    echo "----------------------------------------"
    
    # 检查Maxwell是否已运行
    PIDS=$(pgrep -f "maxwell")
    if [ -n "$PIDS" ]; then
        echo -e "${YELLOW}⚠ Maxwell已在运行 (PID: $PIDS)${NC}"
        echo "停止现有Maxwell进程..."
        kill $PIDS
        sleep 3
    fi
    
    # 检查Maxwell安装
    MAXWELL_HOME="/opt/module/maxwell"
    if [ ! -d "$MAXWELL_HOME" ]; then
        echo -e "${RED}✗ Maxwell未安装或路径不正确: $MAXWELL_HOME${NC}"
        return 1
    fi
    
    # 使用专门的配置文件
    CONFIG_FILE="$MAXWELL_HOME/config_smart_tour.properties"
    if [ ! -f "$CONFIG_FILE" ]; then
        echo -e "${YELLOW}⚠ 配置文件不存在，使用默认配置${NC}"
        CONFIG_FILE="$MAXWELL_HOME/config.properties"
    fi
    
    if [ ! -f "$CONFIG_FILE" ]; then
        echo -e "${RED}✗ 配置文件不存在: $CONFIG_FILE${NC}"
        return 1
    fi
    
    echo "启动Maxwell..."
    echo "配置文件: $CONFIG_FILE"
    
    # 创建日志目录
    mkdir -p "$MAXWELL_HOME/logs"
    
    # 启动Maxwell（后台运行）
    nohup $MAXWELL_HOME/bin/maxwell --config="$CONFIG_FILE" > "$MAXWELL_HOME/logs/maxwell_smart_tour.log" 2>&1 &
    MAXWELL_PID=$!
    
    # 等待启动
    sleep 5
    
    # 检查启动状态
    if kill -0 $MAXWELL_PID 2>/dev/null; then
        echo -e "${GREEN}✓ Maxwell启动成功 (PID: $MAXWELL_PID)${NC}"
        echo "日志文件: $MAXWELL_HOME/logs/maxwell_smart_tour.log"
        return 0
    else
        echo -e "${RED}✗ Maxwell启动失败${NC}"
        echo "请检查日志文件: $MAXWELL_HOME/logs/maxwell_smart_tour.log"
        return 1
    fi
}

# 启动Kafka消费者监控
start_kafka_consumer() {
    echo -e "${BLUE}6. 启动Kafka消费者监控${NC}"
    echo "----------------------------------------"
    
    if command -v kafka-console-consumer.sh >/dev/null 2>&1; then
        echo "启动Kafka消费者监控..."
        echo "监控主题: $KAFKA_TOPIC"
        echo "按 Ctrl+C 停止监控"
        echo "----------------------------------------"
        
        # 启动消费者（后台运行）
        kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic $KAFKA_TOPIC --from-beginning > /tmp/maxwell_smart_tour_output.log 2>&1 &
        CONSUMER_PID=$!
        
        echo "Kafka消费者已启动 (PID: $CONSUMER_PID)"
        echo "输出文件: /tmp/maxwell_smart_tour_output.log"
        
        # 等待消费者启动
        sleep 3
        
        return 0
    else
        echo -e "${YELLOW}⚠ 未找到kafka-console-consumer.sh，跳过Kafka监控${NC}"
        return 1
    fi
}

# 执行数据库操作测试
perform_database_operations() {
    echo -e "${BLUE}7. 执行数据库操作测试${NC}"
    echo "----------------------------------------"
    
    echo "执行smart_tour数据库操作..."
    echo "等待5秒让Maxwell准备就绪..."
    sleep 5
    
    # 执行INSERT操作
    echo "1. 执行INSERT操作..."
    mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
INSERT INTO maxwell_test_table (name, description) VALUES 
('Maxwell测试1', '这是通过Maxwell监控的测试数据1'),
('Maxwell测试2', '这是通过Maxwell监控的测试数据2');
EOF
    
    sleep 2
    
    # 执行UPDATE操作
    echo "2. 执行UPDATE操作..."
    mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
UPDATE maxwell_test_table SET description = '这是更新后的描述', status = 'inactive' WHERE name = 'Maxwell测试1';
EOF
    
    sleep 2
    
    # 执行DELETE操作
    echo "3. 执行DELETE操作..."
    mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
DELETE FROM maxwell_test_table WHERE name = 'Maxwell测试2';
EOF
    
    sleep 2
    
    # 执行DDL操作
    echo "4. 执行DDL操作..."
    mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
ALTER TABLE maxwell_test_table ADD COLUMN priority INT DEFAULT 0;
UPDATE maxwell_test_table SET priority = 1 WHERE name = 'Maxwell测试1';
EOF
    
    echo -e "${GREEN}✓ 数据库操作完成${NC}"
}

# 检查Maxwell输出
check_maxwell_output() {
    echo -e "${BLUE}8. 检查Maxwell输出${NC}"
    echo "----------------------------------------"
    
    echo "等待5秒让Maxwell处理完成..."
    sleep 5
    
    # 检查Maxwell日志
    echo "检查Maxwell日志..."
    if [ -f "/opt/module/maxwell/logs/maxwell_smart_tour.log" ]; then
        echo "Maxwell日志最后20行:"
        echo "----------------------------------------"
        tail -20 /opt/module/maxwell/logs/maxwell_smart_tour.log
        echo "----------------------------------------"
    fi
    
    # 检查Kafka输出
    if [ -f "/tmp/maxwell_smart_tour_output.log" ]; then
        echo "Kafka输出内容:"
        echo "----------------------------------------"
        cat /tmp/maxwell_smart_tour_output.log
        echo "----------------------------------------"
        
        # 分析输出
        echo "分析Maxwell输出..."
        if grep -q "INSERT" /tmp/maxwell_smart_tour_output.log; then
            echo -e "${GREEN}✓ 检测到INSERT操作${NC}"
        fi
        
        if grep -q "UPDATE" /tmp/maxwell_smart_tour_output.log; then
            echo -e "${GREEN}✓ 检测到UPDATE操作${NC}"
        fi
        
        if grep -q "DELETE" /tmp/maxwell_smart_tour_output.log; then
            echo -e "${GREEN}✓ 检测到DELETE操作${NC}"
        fi
        
        if grep -q "ALTER" /tmp/maxwell_smart_tour_output.log; then
            echo -e "${GREEN}✓ 检测到DDL操作${NC}"
        fi
        
        # 检查是否包含smart_tour数据库
        if grep -q "smart_tour" /tmp/maxwell_smart_tour_output.log; then
            echo -e "${GREEN}✓ 检测到smart_tour数据库操作${NC}"
        else
            echo -e "${YELLOW}⚠ 未检测到smart_tour数据库操作${NC}"
        fi
    else
        echo -e "${YELLOW}⚠ 未找到Kafka输出文件${NC}"
    fi
}

# 显示测试结果
show_test_results() {
    echo -e "${BLUE}9. 测试结果总结${NC}"
    echo "----------------------------------------"
    
    echo "smart_tour数据库Maxwell监控测试完成！"
    echo ""
    echo "测试项目:"
    echo "  ✓ smart_tour数据库检查"
    echo "  ✓ Maxwell用户权限检查"
    echo "  ✓ 测试表创建"
    echo "  ✓ Kafka主题检查"
    echo "  ✓ Maxwell启动测试"
    echo "  ✓ 数据库操作测试"
    echo "  ✓ 输出验证测试"
    echo ""
    echo "如果所有测试都通过，说明Maxwell配置正确并能正常捕捉smart_tour数据库变化。"
    echo ""
    echo "后续操作建议:"
    echo "1. 监控Maxwell日志: tail -f /opt/module/maxwell/logs/maxwell_smart_tour.log"
    echo "2. 监控Kafka主题: kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic $KAFKA_TOPIC --from-beginning"
    echo "3. 检查Maxwell状态: ps aux | grep maxwell"
    echo "4. 查看数据库变化: mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD -e 'USE $DATABASE; SELECT * FROM maxwell_test_table;'"
}

# 主函数
main() {
    case "$1" in
        "test")
            # 执行完整测试
            check_smart_tour_database && \
            check_maxwell_permissions && \
            create_test_table && \
            check_kafka_topic && \
            start_maxwell_monitoring && \
            start_kafka_consumer && \
            perform_database_operations && \
            check_maxwell_output && \
            show_test_results
            ;;
        "monitor")
            # 仅启动监控
            start_maxwell_monitoring && \
            start_kafka_consumer
            ;;
        "operations")
            # 仅执行数据库操作
            perform_database_operations
            ;;
        "check")
            # 仅检查环境
            check_smart_tour_database && \
            check_maxwell_permissions && \
            check_kafka_topic
            ;;
        *)
            echo "使用方法: $0 {test|monitor|operations|check}"
            echo ""
            echo "命令说明:"
            echo "  test       - 执行完整测试"
            echo "  monitor    - 启动监控"
            echo "  operations - 执行数据库操作"
            echo "  check      - 检查环境"
            ;;
    esac
}

# 执行主函数
main "$@" 